﻿using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace Sensor.UI.Processors;

public class DataProcessor : IDisposable
{
    private BlockingCollection<byte[]>? _queue;
    private CancellationTokenSource? _cancellationTokenSource;
    private Task? _processorTask;
    private bool _disposedValue;
    

    // 支持异步处理的回调（推荐），若需保持同步可保留Action
    public Func<byte[], CancellationToken,Task>? OnProcessAsync { get; set; }


    /// <summary>
    /// 添加数据到队列
    /// </summary>
    public void Add(byte[] data)
    {
        ArgumentNullException.ThrowIfNull(data);

        if (_cancellationTokenSource == null)
        {
            return;
        }

        try
        {
            _queue?.TryAdd(data,100, _cancellationTokenSource.Token);
        }
        catch (AggregateException ex)
        {
            // 过滤取消异常（正常流程）
            ex.Handle(e => e is OperationCanceledException);
        }
    }


    /// <summary>
    /// 启动数据处理任务
    /// </summary>
    public void Start()
    {
        // 确保前一个任务已停止
        Stop();

        _queue = new BlockingCollection<byte[]>(1024*1024*5);
        _cancellationTokenSource = new CancellationTokenSource();
        var cancellationToken = _cancellationTokenSource.Token;

        _processorTask= Task.Factory.StartNew(async () =>
        {
            await ProcessQueueAsync(cancellationToken).ConfigureAwait(false);

        }, TaskCreationOptions.LongRunning);
        
    }


    /// <summary>
    /// 停止数据处理任务
    /// </summary>
    public void Stop()
    {
        try
        {
            // 取消令牌并等待任务结束
            _cancellationTokenSource?.Cancel();
            if (_processorTask != null && !_processorTask.IsCompleted)
            {
                // 最多等待1秒，避免无限阻塞
                _processorTask.Wait(TimeSpan.FromMilliseconds(100));
            }
        }
        catch (AggregateException ex)
        {
            // 过滤取消异常（正常流程）
            ex.Handle(e => e is OperationCanceledException);
        }
        finally
        {
            // 释放资源
            _processorTask?.Dispose();
            _processorTask = null;
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;

            // 先标记“不再添加元素”，阻止生产者继续添加
            if (_queue!=null)
            {
                _queue.CompleteAdding();
                while (_queue.TryTake(out _, TimeSpan.FromMilliseconds(1)))
                {
                    // 超时100ms是为了避免极端情况下的无限等待（理论上CompleteAdding后不会有新元素）
                } 
            }
            
        }
    }


    /// <summary>
    /// 异步处理队列数据
    /// </summary>
    private async Task ProcessQueueAsync(CancellationToken cancellationToken)
    {
        if (_queue==null)
        {
            return;
        }

        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                if (_queue.TryTake(out var data,-1, cancellationToken))
                {
                    // 优先调用异步回调处理数据
                    if (OnProcessAsync == null)
                    {
                        continue;
                    }
                    await OnProcessAsync(data, cancellationToken).ConfigureAwait(false);
                }

            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                // 正常取消，退出循环
                break;
            }
            catch (Exception)
            {
                // 记录异常（建议使用日志框架）
            }
        }
    }


    #region IDisposable实现
    protected virtual void Dispose(bool disposing)
    {
        if (!_disposedValue)
        {
            if (disposing)
            {
                // 释放托管资源
                Stop();
            }

            _disposedValue = true;
        }
    }

    public void Dispose()
    {
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion
}